<?php

use Workerman\Worker;
use Workerman\Connection\AsyncTcpConnection;

require_once __DIR__ . '/../../vendor/autoload.php';
require_once __DIR__ . '/../../app/config/config.php';

function callbackRecieve($message = [])
{
    if (isset($message['status']) && $message['status'] != HUOBI_SUBSCRIBE_DATA) {
        printf("%s\r\n", json_encode(func_get_args()));
        return;
    }

    static $last_time_stamp = -1;
    static $last_data = [];

    $data = $message['data'];

    // 汇报超过1分钟的数据，
    $tick = $data['tick'];
    if ($tick['id'] > $last_time_stamp && $last_time_stamp > 0) {
        // $params = http_build_query(json_encode($data));
        $params = json_encode($last_data);
        $url = getUrlForKline($data['ch']);

        try {
            list($head, $body) = httpRequest($url, $params, 5);
            printf("%s\r\n%s\r\n", $url, $body);

            echo PHP_EOL;
        } catch (RuntimeException $e) {
            printf("error\t%s\r\n%s\r\n", $url, $e->getMessage());
            Worker::log($e->getMessage());
        }
    }
    // 缓存最后一次回调数据
    $last_time_stamp = $tick['id'];
    $last_data = $data;

    printf("%s=%s,%s\r\n", date('Y-m-d H:i:s', $last_time_stamp), $last_time_stamp, $data['ch']);

}


/**
 * 根据数据包解析协议及状态
 * @param $data
 * @return int
 */
function getSubscribeStatus(&$data)
{
    // WebSocket API 支持双向心跳
    if (isset($data['ping'])) {
        return HUOBI_PING_OK;
    }

    // 订阅成功返回数据
    if (isset($data['ch'])) {
        return HUOBI_SUBSCRIBE_DATA;
    }
    // 订阅成功
    if (isset($data['subbed'])) {
        return HUOBI_SUBSCRIBE_OK;
    }

    if (isset($data['id']) && isset($data['status'])) {
        //错误订阅
        if ($data['status'] == 'error') {
            return HUOBI_SUBSCRIBE_ERROR;
        }
    }

    return HUOBI_PING_ERROR;
}


/*
*订阅数据函数
$sub_str type: string e.g market.btcusdt.kline.1min 具体请查看api
$callback type: function 回调函数，当获得数据时会调用
*/
function subscribe($coin = "btcusdt", $kline = '1min', $callback = 'callbackRecieve')
{
    global $CONIS_HASH;

    if (empty($coin) || !in_array($coin, $CONIS_HASH)) {
        printf("invalid coin : %s .\r\n", $coin);
        exit();
    }

// 暂停片刻
    usleep(mt_rand(10, 1000));

    $GLOBALS['callback'] = $callback;

    $GLOBALS['sub_kline'] = sprintf("market.%s.kline.%s", $coin, $kline);

//    $GLOBALS['sub_depth'] = sprintf("market.%s.depth.step1", $coin);
//    $GLOBALS['sub_trade'] = sprintf("market.%s.trade.detail", $coin);
//    $GLOBALS['sub_detail'] = sprintf("market.%s.detail", $coin);


    $worker_kline = new Worker();
    $worker_kline->onWorkerStart = function ($worker) {

        // ssl需要访问443端口
        $con = new AsyncTcpConnection(API_HUOBI_SUBSCRIBE_WEBSOCKET);

        // 设置以ssl加密方式访问，使之成为wss
        $con->transport = 'ssl';

        $con->onConnect = function ($ws_connnection) {
            $data = json_encode([
                'sub' => $GLOBALS['sub_kline'],
                'id' => $GLOBALS['sub_kline'] . time()
            ]);
            $ws_connnection->send($data);
        };

        $con->onError = function ($ws_connnection, $code, $message) {
            Worker::log($message);
            $ws_connnection->close();
        };
        $con->onClose = function ($ws_connnection) {
            // 如果连接断开，则在1秒后重连
            $ws_connnection->reConnect(5);
        };

        $con->onMessage = function ($ws_connnection, $data) {
            $data = gzdecode($data);
            $data = json_decode($data, true);

            $status = getSubscribeStatus($data);
            if ($status == HUOBI_PING_OK) {
                $ws_connnection->send(json_encode(["pong" => $data['ping']]));
                return;
            }
            // 回调
            call_user_func_array($GLOBALS['callback'], array(['name' => 'kline', 'status' => $status, 'data' => $data]));

        };

        $con->connect();
    };

    Worker::runAll();
}

// __MAIN__

$path = pathinfo($argv[0]);
$list = explode('.', $path['basename']);

$coin = $list[0];
$kline = $list[1];
// 启动
subscribe(strtolower($coin), strtolower($kline));
